package com.heima.article.listener;

import com.alibaba.fastjson.JSON;
import com.heima.article.dto.ArticleStreamMessage;
import com.heima.article.entity.UpdateArticleMessage;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.*;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.cloud.stream.annotation.EnableBinding;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.messaging.handler.annotation.SendTo;
import org.springframework.util.StringUtils;

import java.time.Duration;

@EnableBinding(value = ArticleProcess.class)
public class ArticleListener {

    //时间
    @Value("${commit.time}")
    private String commitTime;

    //数据来源主题
    @StreamListener(value = "article_behavior")
    //数据结果发往那个主题
    @SendTo(value = "article_result")
    public KStream<String, String> process(KStream<String, String> input) {
        //接收消息的格式为:UpdateArticleMessage
        KStream<String, String> map = input.map(new KeyValueMapper<String, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(String key, String value) {
                // value 是json = {"articleId":1540597913363701761,"type":1,"add":1}

                System.out.println("接收到的消息:"+value);
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
               //提取文章id
                Long articleId = updateArticleMessage.getArticleId();
                //返回键值对
                return new KeyValue<>(articleId.toString(), value);
            }
        });
        //以每一篇文章id进行分组
        KGroupedStream<String, String> groupBy = map.groupByKey();

        //统计时间窗口数据
        TimeWindowedKStream<String, String> windowedBy = groupBy.windowedBy(TimeWindows.of(Duration.ofMillis(Long.parseLong(commitTime))));
        //初始化聚合结果
        Initializer<String> init = new Initializer<String>() {
            @Override
            public String apply() {
                System.out.println("聚合初始化....");
                return null;
            }
        };
        //聚合结果的逻辑
        Aggregator<String, String, String> aggregator = new Aggregator<String, String, String>() {
            @Override
            public String apply(String key, String value, String aggregate) {
                //每次接收到新消息,都会在这个方法里面执行一次
                //key  ---->  文章id
                //value ---->  json={"articleId":1540597913363701761,"type":1,"add":1}
                // aggregate 是在同一个时间窗口内上一次聚合的结果
                // 聚合处理的结果是 ArticleStreamMessage
                System.out.println("开始本次消息的处理.....");
                System.out.println("上次聚合的结果:"+aggregate);
                ArticleStreamMessage message = null;
                //判断上一次聚合结果是否为空
                if (StringUtils.isEmpty(aggregate)) {
                    //新建
                    message = new ArticleStreamMessage();
                    message.setArticleId(Long.parseLong(key));
                } else {
                    //如果不为空,结果直接从上一次提取
                    message = JSON.parseObject(aggregate, ArticleStreamMessage.class);
                }
                //提取本次接受的消息
                UpdateArticleMessage updateArticleMessage = JSON.parseObject(value, UpdateArticleMessage.class);
                //处理本次消息    操作类型 0 阅读 1 点赞 2 评论 3 收藏
                switch (updateArticleMessage.getType()){
                    case 0:
                        // 0 阅读 将message中的阅读数量加上本次的增量
                        message.setView(message.getView()+updateArticleMessage.getAdd());
                        System.out.println("阅读量增加: " + updateArticleMessage.getAdd());
                        break;
                    case 1:
                        // 1 点赞 将message中的点赞数量加上本次的增量
                        message.setLike(message.getLike()+ updateArticleMessage.getAdd());
                        System.out.println("点赞量增加: " + updateArticleMessage.getAdd());
                        break;
                    case 2:
                        // 2 评论 将message中的评论数量加上本次的增量
                        message.setComment(message.getComment()+updateArticleMessage.getAdd());
                        System.out.println("评论量增加: " + updateArticleMessage.getAdd());
                        break;
                    case 3:
                        // 3 收藏 将message中的收藏数量加上本次的增量
                        message.setCollect(message.getCollect()+updateArticleMessage.getAdd());
                        System.out.println("收藏量增加: " + updateArticleMessage.getAdd());
                        break;
                }
                //将本次更新结果保存到聚合中间结果
                String json = JSON.toJSONString(message);
                System.out.println("本次聚合完的结果:"+json);
                return json;
            }
        };
        //获取聚合结果
        KTable<Windowed<String>, String> aggregate = windowedBy.aggregate(init, aggregator);
       //转换聚合结果
        KStream<String, String> stream = aggregate.toStream().map(new KeyValueMapper<Windowed<String>, String, KeyValue<String, String>>() {
            @Override
            public KeyValue<String, String> apply(Windowed<String> key, String value) {

                System.out.println("时间窗口的最终执行结果:"+value);
                return new KeyValue<>(key.key(), value);
            }
        });
        // 最终发送到结果的value是 ArticleStreamMessage转换成json格式
        return stream;
    }
}
